package com.atguigu.flink.datastreamapi.transform;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by Smexy on 2023/4/4
 *
 *SingleOutputStreamOperator.process(ProcessFunction<T,O>)
 *      T:输入类型
 *      O：输出类型
 *
 *
 *      KeyedStream.process(KeyedProcessFunction<K, I, O>)
 *          K: key的类型
 *          I： 输入
 *          O： 输出
 *
 *
 */
public class Demo3_ProcessKeySum
{
    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.setParallelism(2);

        SingleOutputStreamOperator<WaterSensor> ds = env.socketTextStream("hadoop102", 8888)
                                                               .map(new WaterSensorMapFunction());

        //模拟process 按照id对传感器进行vc累加
        KeyedStream<WaterSensor, String> ds2 = ds.keyBy(WaterSensor::getId);

        ds2
          .process(new KeyedProcessFunction<String, WaterSensor, String>()
          {
              //同一种key一定会发送到下游的同一个 process实例中，不同的key也能发送到同一个process实例。
              //如何解决：  后续可以通过状态编程解决(flink的优势)
              // 临时解决:   维护Map<String,Integer> sumVc
              private Map<String,Integer> vcMap = new HashMap<>();
              // ctx.getCurrentKey(): 获取当前的类，一般keyBy之后是可以获取。
              @Override
              public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                  String key = ctx.getCurrentKey();
                  //从Map中获取当前key累积的结果
                  Integer preSumVc = vcMap.getOrDefault(key, 0);
                  preSumVc += value.getVc();
                  vcMap.put(key,preSumVc);
                  out.collect(key+" 累积 sumVc :"+preSumVc);
              }
          })
          .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
